#include "threadDefs.h"
#include "ProcessThread.h"
#include "OutThread.h"

ProcessThread::ProcessThread(boost::asio::io_service& io) :
    inQueue(io, 180000, &ProcessThread::inQueueHasDataWrapper, this)
    {
    bInQueueDataReady = false;
    inQueue.startWork();
    }

ProcessThread::~ProcessThread()
    {
    }

void ProcessThread::operator()()
    {
    while (true)
        {
        boost::unique_lock<boost::mutex> lock(mut);
        while (!bInQueueDataReady)
            {
            cond.wait(lock);
            }
        processInQueueData();
        }

    }

void ProcessThread::processInQueueData()
    {
    Message * pMsg = NULL;

    while (inQueue.removeRecord(pMsg) == true)
        {
        //inQueue still has record
        processQueue(pMsg);
        }
    }

void ProcessThread::inQueueHasDataWrapper(void * pObj)
    {
    return reinterpret_cast<ProcessThread *> (pObj)->inQueueHasData();
    }

void ProcessThread::inQueueHasData()
    {
    boost::lock_guard<boost::mutex> lock(mut);
    bInQueueDataReady = true;
    cond.notify_one();
    }

void ProcessThread::processQueue(Message * pMsg)
    {
    if (pMsg)
        {
        processMsg(pMsg);
        LOGCON.infoStream() << *pMsg;
        if (pOutThread && pOutThread->getInQueue().insertRecord(pMsg, false))
            {
                //Let the last dealer to free pMsg.
                LOGCON.infoStream() << __PRETTY_FUNCTION__
                        << " - OK to insert\n";
            }
        else
            {
            LOGCON.errorStream() << __PRETTY_FUNCTION__
                    << " - Fail to insert\n";
            delete pMsg;
            }

        }

    }

void ProcessThread::processMsg(Message * pMsg)
    {
    char * tag = "-ProcessThread";

    strcat((char*) (pMsg->body()), tag);
    pMsg->body_length(strlen((char*) (pMsg->body())) + 1);
    pMsg->encode_header();
    }
